package org.zoomdev.zoom.dao.transaction;


import org.zoomdev.zoom.common.io.Io;
import org.zoomdev.zoom.dao.Ar;
import org.zoomdev.zoom.dao.ConnectionExecutor;
import org.zoomdev.zoom.dao.Dao;
import org.zoomdev.zoom.dao.DaoException;
import org.zoomdev.zoom.dao.impl.ArImpl;

import javax.sql.DataSource;
import java.sql.Blob;
import java.sql.Clob;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

public class Transactions {

    /**
     * 共享数据
     */
    private static final DataSource[] dataSources;
    private static final AtomicInteger counter = new AtomicInteger();
    private static final ThreadLocal<Transactions> threadLocal = new ThreadLocal<>();

    /**
     * 总共支持10个数据库连接
     */
    static {
        dataSources = new DataSource[10];
    }


    /**
     * 线程私有数据
     */


    /**
     * 正在进行事务处理
     */
    private final Set<Transaction> transacting = new HashSet<>();
    private final Ar[] records = new Ar[10];
    private int level;
    private int ref;

    public static int register(Dao dao) {
        int index = counter.getAndIncrement();
        dataSources[index] = dao.dataSource();
        return index;
    }

    public static Transactions get() {
        Transactions transactions = threadLocal.get();
        if (transactions == null) {
            transactions = new Transactions();
            threadLocal.set(transactions);
        }
        return transactions;
    }

    public static void trans(Runnable runnable) {
        Transactions transactions = Transactions.get();
        try {
            transactions.beginTrans(Trans.TRANSACTION_READ_COMMITTED);
            runnable.run();
            transactions.commitTrans();
        } catch (Throwable t) {
            transactions.rollbackTrans();
        }
    }

    /**
     * 将数据库中的类型转为一般常用类型
     *
     * @param dataType
     * @return
     */
    public static Class<?> normalizeType(Class<?> dataType) {
        if (Clob.class.isAssignableFrom(dataType)) {
            return String.class;
        }
        if (Blob.class.isAssignableFrom(dataType)) {
            return byte[].class;
        }
        return dataType;
    }

    /**
     * 释放
     */
    public void release() {
        threadLocal.remove();
    }

    public Connection getConnection(DataSource ds) throws SQLException {
        if (isTrans()) {
            for (Transaction trans : transacting) {
                if (trans.ds == ds) {
                    return trans.getConnection();
                }
            }
            Transaction trans = new Transaction(level, ds);
            transacting.add(trans);
            return trans.getConnection();
        }
        return ds.getConnection();
    }

    public Ar getActiveRecord(int index) {
        return records[index];
    }

    public Ar setActiveRecord(int index, ArImpl ar) {
        records[index] = ar;
        return ar;
    }

    void addRef() {
        ++ref;
    }

    boolean subRef() {
        --ref;
        return ref <= 0;
    }

    public void beginTrans(int level) {
        if (isTrans()) {
            addRef();
        } else {
            this.level = level;
            this.ref = 1;
        }
    }

    public void commitTrans() {
        if (subRef()) {
            try {
                for (Transaction transaction : transacting) {
                    transaction.commit();
                }
            } finally {
                transacting.clear();
            }
        }
    }

    public void rollbackTrans() {
        if (subRef()) {
            for (Transaction transaction : transacting) {
                transaction.rollback();
            }
            transacting.clear();
        }
    }

    public boolean isTrans() {
        return ref > 0;
    }

    public int getLevel() {
        return level;
    }

    public void releaseConnection(Connection connection) {
        if (!isTrans()) {
            Io.close(connection);
        }
    }

    public <T> T execute(DataSource ds, ConnectionExecutor executor) {
        Connection connection = null;
        try {
            connection = getConnection(ds);
            return (T) executor.execute(connection);
        } catch (SQLException e) {
            throw new DaoException(e);
        } finally {
            releaseConnection(connection);
        }
    }


}
